package com.elinshaw.wordcount;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.Map;

/**
 * @author Administrator
 */
public class PrintBolt extends BaseRichBolt {

    private static final long serialVersionUID = -1512537746316594950L;

    @Override
    public void prepare(Map map, TopologyContext topologycontext, OutputCollector outputcollector) {

    }

    @Override
    public void execute(Tuple tuple) {
        String word = tuple.getStringByField("word");
        int count = tuple.getIntegerByField("count");
        System.out.println(word + "-" + count);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputfieldsdeclarer) {

    }

    /**
     * 在bolt被关闭的时候调用， 它应该清理所有被打开的资源。但是集群不保证这个方法一定会被执行。比如执行task的机器down掉了，那么根本就没有办法来调用那个方法。
     */
    @Override
    public void cleanup() {
        System.out.println("---------- FINAL COUNTS -----------");
        System.out.println("----------------------------");
    }
}  